package com.example.demo.service.impl;

import cn.hutool.core.util.IdUtil;
import com.example.demo.bean.MessageBody;
import com.example.demo.enums.MsgEnum;
import com.example.demo.exception.CustomException;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;


@Slf4j
@Component
public class MQServiceImpl {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    private void sendMsg(MsgEnum msg_type,String destination,Object payload,Integer merchantId,String msgSource){
        String msgKey = IdUtil.simpleUUID();
        MessageBody msgBody = new MessageBody(msgKey, payload , msgSource);
        Message<MessageBody> message = MessageBuilder.withPayload(msgBody).setHeader("KEYS",msgKey ).build();
        log.info(String.format("消息发送 MQService 开始: %s %s", destination, message));
        SendResult result = null;
        switch (msg_type) {
            case ONE_WAY:
                rocketMQTemplate.sendOneWay(destination, message);
                break;
            case ASYNC:
                //todo hashkey设置成商家id
                if(merchantId == -1 ){
                    rocketMQTemplate.asyncSend(destination, message,new SendCallback() {
                        @Override
                        public void onSuccess(SendResult sendResult) {
                        }
                        @Override
                        public void onException(Throwable throwable) {
                            log.error("MQService:" + ExceptionUtils.getStackTrace(throwable));
                            throw new CustomException("消息发送失败 topic_tag:" + destination);
                        }
                    });
                }else {
                    rocketMQTemplate.asyncSendOrderly(destination, message,merchantId+"",new SendCallback() {
                        @Override
                        public void onSuccess(SendResult sendResult) {
                        }
                        @Override
                        public void onException(Throwable throwable) {
                            log.error("MQService:" + ExceptionUtils.getStackTrace(throwable));
                            throw new CustomException("消息发送失败 topic_tag:" + destination);
                        }
                    });
                }
                break;
            case SYNC:
                result = rocketMQTemplate.syncSend(destination, message);
                break;
        }
        log.info(String.format("消息发送 MQService 结束: msgId: %s dest: %s msg: %s",result != null ? result.getMsgId() : "", destination, message));
    }

    /**
     * 同步发送消息,会确认应答
     * @param destination
     * @param payload
     */
    public void syncSendMsg(String destination, Object payload, String msgSource){
        sendMsg(MsgEnum.SYNC,destination, payload,-1,msgSource) ;
    }
    /**
     * 同步发送消息,会确认应答
     * @param topic
     * @param tag
     * @param payload
     */
    public void syncSendMsg(String topic, String tag, Object payload, String msgSource){
        // 发送的消息体，消息体必须存在
        // 业务主键作为消息key
        String destination = topic + ":" + tag;
        syncSendMsg(destination, payload,msgSource);
    }

    /**
     * 异步消息发送,异步日志确认异常
     * @param destination
     * @param payload
     */
    public void asyncSendMsg(String destination, Object payload, String msgSource){
        sendMsg(MsgEnum.ASYNC,destination, payload,-1,msgSource);
    }

    public void asyncSendOrderlyMsg(String destination, Object payload,int merchantId,String msgSource){
        sendMsg(MsgEnum.ASYNC,destination, payload,merchantId,msgSource);
    }
    /**
     * 异步消息发送,异步日志确认异常
     * @param topic
     * @param tag
     * @param payload
     * @return
     */
    public void asyncSendMsg(String topic, String tag, Object payload, String msgSource){
        // 发送的消息体，消息体必须存在
        // 业务主键作为消息key
        String destination = topic + ":" + tag;
        asyncSendMsg(destination, payload,msgSource);
    }

    /**
     * 单向发送消息，不关注结果
     * @param destination
     * @param payload
     */
    public void oneWaySendMsg(String destination, Object payload, String msgSource){
        sendMsg(MsgEnum.ONE_WAY,destination, payload,-1,msgSource);
    }
    /**
     * 单向发送消息，不关注结果
     * @param topic
     * @param tag
     * @param payload
     */
    public void oneWaySendMsg(String topic, String tag, Object payload, String msgSource){
        // 发送的消息体，消息体必须存在
        // 业务主键作为消息key
        String destination = topic + ":" + tag;
        oneWaySendMsg(destination, payload,msgSource);
    }
}
